package org.xq.mq.listener;

import org.xq.mq.component.SpringContextHolder;
import org.xq.mq.constant.Constants;
import org.xq.mq.container.RedisQueueConsumerContainer;
import org.xq.mq.model.RedisQueueMessage;
import org.xq.mq.service.RedisQueueConsumer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RedisQueueRunnable implements Runnable {

    @Resource
    private RedisQueueConsumer redisQueueConsumer;

    RedisTemplate redisTemplate;

    public RedisQueueRunnable(RedisQueueConsumer redisQueueConsumer) {
        this.redisQueueConsumer = redisQueueConsumer;
    }

    @Override
    public void run() {
        redisTemplate = (RedisTemplate) SpringContextHolder.getBean("redisTemplate");

        String queueName = "queue:"+redisQueueConsumer.getQueueName();
        log.info("redis监听器开始监听:{}", queueName);

        BoundListOperations boundListOperations = redisTemplate.boundListOps(queueName);

        List range = boundListOperations.range(0, -1);
        while (RedisQueueConsumerContainer.isRun && range!=null) {
            try {
                Object obj = boundListOperations.rightPop(
                        Constants.REDIQUEUELISTENER_LPOP_WAIT_TIME,
                        TimeUnit.SECONDS);
                /**
                 * 如果消息队列中有消息,则进行消费
                 */
                JSONObject jsonObject = JSON.parseObject((String) obj);
                if (obj != null) {
                    redisQueueConsumer.getMessage(new RedisQueueMessage(jsonObject.getString("content")));
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.info("redis监听器错误：{}", redisQueueConsumer.getQueueName());
            }
        }
    }
}
